package com.shujia.sql

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

object Demo2FLinkSqlOnKafka {
  def main(args: Array[String]): Unit = {


    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink的计划器
      .inStreamingMode() //使用流模型
      .build()

    //窗口table 环境
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    val configuration = new Configuration()
    //如果主键为null,自动删除
    configuration.setString("table.exec.sink.not-null-enforcer", "drop")
    bsTableEnv.getConfig.addConfiguration(configuration)

    /**
      * 基于kafka的topic创建动态表
      *
      */

    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE student (
        | id STRING,
        | name STRING,
        | age BIGINT,
        | gender STRING,
        | clazz STRING
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'student',
        | 'properties.bootstrap.servers' = 'master:9092',
        | 'properties.group.id' = 'asdasdasd',
        | 'format' = 'csv',
        | 'scan.startup.mode' = 'latest-offset',
        | 'csv.field-delimiter'=',',
        | 'csv.ignore-parse-errors' ='true'
        |)
        |
      """.stripMargin)


    /**
      * 用于测试的打印表
      *
      */

    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE print_table (
        | clazz STRING,
        | c BIGINT
        |) WITH (
        | 'connector' = 'print'
        |)
        |
      """.stripMargin)

    /**
      * 需要现在mysql中创建clazz_num表
      * 字段名和字段类型要一致
      *
      */

    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE mysql_clazz_num (
        |  clazz STRING,
        |  num BIGINT,
        |  PRIMARY KEY (clazz) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/student?useUnicode=true&characterEncoding=utf-8',
        |   'table-name' = 'clazz_num',
        |   'username' ='root',
        |   'password'='123456'
        |)
        |
      """.stripMargin)


    /**
      * 执行sql查询，将查询的结果保存到另一张表中
      *
      */
    bsTableEnv.executeSql(
      """
        |insert into mysql_clazz_num
        |select clazz,count(1) as num from student group by clazz
        |
      """.stripMargin)


    /**
      *
      * executeSql: 可以执行DDl ,DML DQL语句  不会返回Table对象
      * sqlQuery: 只能执行DQl 语句，会返回一个Table对象
      */
  }

}
